#![allow(unused_variables)]

    use std::collections::HashMap ;
    use serde::{Deserialize, Serialize};
    use serde_json::Value ;
    use serde_with::skip_serializing_none ;
    use reqwest::Client ;
    use async_trait::async_trait ;


    use crate::kafka::{self, KafkaClient} ;
    use crate::kafka::types::* ;

    mod utils {
        use chrono::prelude::*;
        pub fn uuid() -> String {
            uuid::Uuid::new_v4().to_string()
        }
        pub fn now() -> String {

            let utc: DateTime<Utc> = Utc::now();
            utc.format("%Y-%m-%dT%H:%M:%SZ").to_string()
        }
    }

    mod consts {
        pub const VERSION: &str = "2018-10-15" ;
        pub const SIGNATURE_METHOD: &str = "HMAC-SHA1" ;
        pub const SIGNATURE_VERSION: &str = "1.0" ;
        pub const FORMAT: &str = "json" ;
    }

    inventory::submit! {
        kafka::KafkaPlugin {
            new_client: |bootstrap_server: &str| {
                if !bootstrap_server.starts_with("http://alikafka.") { return None };
                let access_key = &dotenv::var("ALIYUN_KAFKA_ACCESS_KEY").expect("ENV:ALIYUN_KAFKA_ACCESS_KEY NOT FIND!");
                let access_secret = &dotenv::var("ALIYUN_KAFKA_ACCESS_SECRET").expect("ENV:ALIYUN_KAFKA_ACCESS_SECRET NOT FIND");
                Some(Box::new(AliyunKafkaClient::new(bootstrap_server, access_key, access_secret)))
            }
        }
    }
    #[derive(Default)]
    pub struct AliyunKafkaClient {
        client: Client,
        bootstrap_server: String,
        access_key: String,
        access_secret: String,
    }
    impl AliyunKafkaClient {
        pub fn new(bootstrap_server: &str, access_key: &str, access_secret: &str) -> Self {
            AliyunKafkaClient {
                bootstrap_server: bootstrap_server.to_string(),
                client: Client::new(),
                access_key: access_key.to_string(),
                access_secret: access_secret.to_string(),
            }
        }
        pub fn url(&self) -> &str {
            self.bootstrap_server.split('#').collect::<Vec<_>>()[0]
        }
        pub fn instance_id(&self) -> &str {
            self.bootstrap_server.split('#').collect::<Vec<_>>()[1]
        }
        pub fn region_id(&self) -> &str {
            self.bootstrap_server.split('.').collect::<Vec<_>>()[1]
        }
        pub fn request_common(&self) -> AliyunRequestCommon {
            AliyunRequestCommon {
                version: consts::VERSION,
                signature_method: consts::SIGNATURE_METHOD,
                signature_version: consts::SIGNATURE_VERSION,
                format: consts::FORMAT,

                region_id: self.region_id(),
                instance_id: self.instance_id(),
                access_key: &self.access_key,
                timestamp: utils::now(),
                signature_nonce: utils::uuid(),
            }
        }
        pub fn sign(&self, params: &mut Vec<(String, String)>)  -> anyhow::Result<()> {
            fn encode(s: &str) -> String {
                urlencoding::encode(s).to_string()
            }
            let query_string = & params.iter().map(|(k, v)|  {format!("{}={}", encode(k), encode(v))}).collect::<Vec<_>>().join("&");
            let sign_string = ["POST", &encode("/"), &encode(query_string)].join("&") ;

            log::debug!("sign_string: {}", sign_string) ;
            use ring::hmac::{sign, Key, HMAC_SHA1_FOR_LEGACY_USE_ONLY} ;
            let key = Key::new(HMAC_SHA1_FOR_LEGACY_USE_ONLY, format!("{}&", self.access_secret).as_bytes()) ;
            let sign_s = base64::encode(sign(&key, sign_string.as_bytes())) ;
            params.push(("Signature".to_string(), sign_s)) ;
            Ok(())
        }
    }

    #[skip_serializing_none]
    #[derive(Serialize)]
    #[serde(rename_all = "PascalCase")]
    pub struct AliyunRequestCommon<'a> {
        version: &'a str,
        region_id: &'a str,
        instance_id: &'a str,
        access_key: &'a str,
        timestamp: String,
        signature_method: &'a str,
        signature_version: &'a str,
        signature_nonce: String,
        format: &'a str,
    }

    fn uppercase_first_letter(s: &str) -> String {
        let mut c = s.chars();
        match c.next() {
            None => String::new(),
            Some(f) => f.to_uppercase().collect::<String>() + c.as_str(),
        }
    }
    #[async_trait]
    impl KafkaClient for AliyunKafkaClient {
        async fn create_topic(&self, req: CreateTopicReq) -> anyhow::Result<Value> {
            #[skip_serializing_none]
            #[derive(Serialize)]
            #[serde(rename_all = "PascalCase")]
            struct AliyunParams<'a> {
                #[serde(flatten)]
                common: AliyunRequestCommon<'a>,
                action: &'a str,
                topic: &'a str,
                remark: &'a str,
                partition_num: &'a Option<String>,
                compact_topic: String,
            }

            let aliyun_request = &AliyunParams {
                common: self.request_common(),
                action: "CreateTopic",
                topic: &req.topic,
                remark: &req.topic,
                partition_num: &req.partitions.map(|x| x.to_string()),
                compact_topic: (req.config.get("cleanup.policy") == Some(&"compact".to_string())).to_string()
            } ;
            let aliyun_request_json = serde_json::to_value(aliyun_request).unwrap() ;
            log::debug!("aliyun_request_json: {}", aliyun_request_json) ;
            let mut params = serde_json::from_value::<HashMap<String, Option<String>>>(aliyun_request_json)?.into_iter()
                .filter_map(|(k, v_o)| { v_o.map(|v|(k, v)) } ).collect::<Vec<(_, _)>>() ;
            params.sort() ;

            self.sign(&mut params)? ;
            log::debug!("post-json: {:?}", params) ;
            Ok(self.client.post(self.url()).form(&params).send().await?.json().await?)
        }
        async fn delete_topic(&self, req: DeleteTopicReq) -> anyhow::Result<Value> {
            #[skip_serializing_none]
            #[derive(Serialize)]
            #[serde(rename_all = "PascalCase")]
            struct AliyunParams<'a> {
                #[serde(flatten)]
                common: AliyunRequestCommon<'a>,
                action: &'a str,
                topic: &'a str,
            }

            let aliyun_request = &AliyunParams {
                common: self.request_common(),
                action: "DeleteTopic",
                topic: &req.topic,
            } ;
            let aliyun_request_json = serde_json::to_value(aliyun_request).unwrap() ;
            let mut params = serde_json::from_value::<HashMap<String, Option<String>>>(aliyun_request_json)?.into_iter()
                .filter_map(|(k, v_o)| {
                    v_o.map(|v|(uppercase_first_letter(&k), v))
                }).collect::<Vec<(_, _)>>() ;
            params.sort() ;

            self.sign(&mut params)? ;
            log::debug!("post-json: {:?}", params) ;
            Ok(self.client.post(self.url()).form(&params).send().await?.json().await?)
        }
        async fn list_topic(&self, req: ListTopicReq) -> anyhow::Result<Value> {
            #[skip_serializing_none]
            #[derive(Serialize)]
            #[serde(rename_all = "PascalCase")]
            struct AliyunParams<'a> {
                #[serde(flatten)]
                common: AliyunRequestCommon<'a>,
                action: &'a str,
            }

            let aliyun_request = &AliyunParams {
                common: self.request_common(),
                action: "GetTopicList",
            } ;
            let aliyun_request_json = serde_json::to_value(aliyun_request).unwrap() ;
            let mut params = serde_json::from_value::<HashMap<String, Option<String>>>(aliyun_request_json)?.into_iter()
                .filter_map(|(k, v_o)| {
                    v_o.map(|v|(uppercase_first_letter(&k), v))
                }).collect::<Vec<(_, _)>>() ;
            params.sort() ;

            self.sign(&mut params)? ;
            log::debug!("post-json: {:?}", params) ;
            let mut resp = self.client.post(self.url()).form(&params).send().await?.json::<Value>().await? ;

            #[derive(Deserialize, Serialize)]
            #[serde(rename_all = "PascalCase")]
            struct TopicDisplay {
                compact_topic: bool,
                create_time: u64,
                // instance_id: String,
                // local_topic: String,
                partition_num: u64,
                // region_id: String,
                // remark: String,
                // status: u64,
                status_name: String,
                topic: String,
            }
            log::debug!("list-resp: {}", resp) ;
            let display = serde_json::from_value::<Vec<TopicDisplay>>(resp["TopicList"]["TopicVO"].take()).unwrap() ;
            Ok(serde_json::to_value(display).unwrap())
        }

        async fn create_consume_group(&self, req: CreateConsumeGroupReq) -> anyhow::Result<Value> {
            #[skip_serializing_none]
            #[derive(Serialize)]
            #[serde(rename_all = "PascalCase")]
            struct AliyunParams<'a> {
                #[serde(flatten)]
                common: AliyunRequestCommon<'a>,
                action: &'a str,
                consumer_id: &'a str,
            }

            let aliyun_request = &AliyunParams {
                common: self.request_common(),
                action: "CreateConsumerGroup",
                consumer_id: &req.group,
            } ;
            let aliyun_request_json = serde_json::to_value(aliyun_request).unwrap() ;
            let mut params = serde_json::from_value::<HashMap<String, Option<String>>>(aliyun_request_json)?.into_iter()
                .filter_map(|(k, v_o)| { v_o.map(|v|(k, v))}).collect::<Vec<(_, _)>>() ;
            params.sort() ;

            self.sign(&mut params)? ;
            log::debug!("post-json: {:?}", params) ;
            let resp = self.client.post(self.url()).form(&params).send().await?.json::<Value>().await? ;
            Ok(resp)
        }
        async fn delete_consume_group(&self, req: DeleteConsumeGroupReq) -> anyhow::Result<Value> {
            #[skip_serializing_none]
            #[derive(Serialize)]
            #[serde(rename_all = "PascalCase")]
            struct AliyunParams<'a> {
                #[serde(flatten)]
                common: AliyunRequestCommon<'a>,
                action: &'a str,
                consumer_id: &'a str,
            }

            let aliyun_request = &AliyunParams {
                common: self.request_common(),
                action: "DeleteConsumerGroup",
                consumer_id: &req.group,
            } ;
            let aliyun_request_json = serde_json::to_value(aliyun_request).unwrap() ;
            let mut params = serde_json::from_value::<HashMap<String, Option<String>>>(aliyun_request_json)?.into_iter()
                .filter_map(|(k, v_o)| {v_o.map(|v|(k, v))}).collect::<Vec<(_, _)>>() ;
            params.sort() ;

            self.sign(&mut params)? ;
            log::debug!("post-json: {:?}", params) ;
            Ok(self.client.post(self.url()).form(&params).send().await?.json().await?)
        }
        async fn list_consume_group(&self, req: ListConsumeGroupReq) -> anyhow::Result<Value> {
            #[skip_serializing_none]
            #[derive(Serialize)]
            #[serde(rename_all = "PascalCase")]
            struct AliyunParams<'a> {
                #[serde(flatten)]
                common: AliyunRequestCommon<'a>,
                action: &'a str,
            }

            let aliyun_request = &AliyunParams {
                common: self.request_common(),
                action: "GetConsumerList",
            } ;
            let aliyun_request_json = serde_json::to_value(aliyun_request).unwrap() ;
            let mut params = serde_json::from_value::<HashMap<String, Option<String>>>(aliyun_request_json)?.into_iter()
                .filter_map(|(k, v_o)| {v_o.map(|v|(k, v))}).collect::<Vec<(_, _)>>() ;
            params.sort() ;

            self.sign(&mut params)? ;
            log::debug!("post-json: {:?}", params) ;
            let mut resp = self.client.post(self.url()).form(&params).send().await?.json::<Value>().await? ;
            Ok(resp["ConsumerList"]["ConsumerVO"].take())
        }
    }
